/*
 * =====================================================================================
 *
 *       Filename:  file_trans_redis.cc
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  2013年06月17日 15时07分22秒
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Xie Yanping (), 
 *        Company:  www.Qihoo.net
 *
 * =====================================================================================
 */

#include "file_trans_redis.h"
#include <boost/shared_ptr.hpp>
#include <boost/format.hpp>
#include <thrift/Thrift.h>
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/transport/TBufferTransports.h>
#include "redis_executor/simple_redis_locator.h"
#include "redis_executor/redis_executor.h"
#include "redis_executor/scoped_reply.h"
#include "base/config_reader.h"

using apache::thrift::transport::TTransport;
using apache::thrift::transport::TMemoryBuffer;
using apache::thrift::protocol::TProtocol;
using apache::thrift::protocol::TBinaryProtocol;

typedef boost::shared_ptr<TTransport> Transport_Ptr;
typedef boost::shared_ptr<TProtocol> Protocol_Ptr;

namespace imserver {

FileTransRedis::FileTransRedis() : redis_exec_(NULL) {
  ConfigReader config("../conf/file_transfer_service.conf");
  RedisLocator *locator = new FixedLocator(
      config.Get("redis", "server_list").c_str());
  redis_exec_.reset(new RedisExecutor(locator));
}

int FileTransRedis::GetFileContent(FileContent& content,
                                   const std::string& identify) {
  std::string redis_command = boost::str(
      boost::format("GET im:filecontent:%1%") % identify);
  ScopedReply reply(redis_exec_->Execute("", redis_command.c_str()));

  if (reply && (reply->type == REDIS_REPLY_STRING)) {
    Transport_Ptr transport(new TMemoryBuffer((uint8_t*)reply->str, reply->len));
    Protocol_Ptr protocol(new TBinaryProtocol(transport));
    content.read(protocol.get());
    LOG_DEBUG("identify=" << content.identify <<
        " content_length=" << content.content_length);
    return 0;
  }

  LOG_ERROR("read file comtent error.(get redis reply failed," <<
      " identify = " << identify << ")");
  return -1;
}

int FileTransRedis::GetFileStream(std::string& body,
                                  const std::string& identify,
                                  const int64_t offset) {
  std::string redis_command = boost::str(
      boost::format("GETRANGE im:file:%1% %2% -1") % identify % offset);
  ScopedReply reply(redis_exec_->Execute("", redis_command.c_str()));

  if (reply && (reply->type == REDIS_REPLY_STRING)) {
    body.append(reply->str, reply->len);
    return 0;
  }

  LOG_ERROR("read file stream error.(get redis reply failed," <<
      " identify = " << identify << ", offset = " << offset << ")");
  return -1;
}

int FileTransRedis::SetFileContent(const FileContent& content) {
  std::string redis_command = boost::str(
      boost::format("SETEX im:filecontent:%1% 86400 %%b") % content.identify);
  Transport_Ptr transport(new TMemoryBuffer(16));
  Protocol_Ptr protocol(new TBinaryProtocol(transport));
  uint32_t size = content.write(protocol.get());
  uint8_t *serialize = NULL;
  (dynamic_cast<TMemoryBuffer*>(transport.get()))->getBuffer(&serialize, &size);
  ScopedReply reply(redis_exec_->Execute("", redis_command.c_str(), serialize, size));

  if (reply &&
      reply->type == REDIS_REPLY_STATUS &&
      strcasecmp(reply->str, "ok") == 0) {
    return 0;
  }

  LOG_ERROR("set file comtent error.");
  return -1;
}

int FileTransRedis::SetFileStream(const std::string& body,
                                  const std::string& identify) {
  redisContext *context = redis_exec_->AllocContext("");
  if (!context) {
    LOG_ERROR("set file stream error.(redis context null," <<
        " identity = " << identify << ")");
    return -1;
  }

  std::string append = boost::str(
      boost::format("APPEND im:file:%1% %%b") % identify);
  std::string expire = boost::str(
      boost::format("EXPIRE im:file:%1% 86400") % identify);//设置一天的超时时间
  if (0 != redisAppendCommand(context, append.c_str(), body.c_str(), body.size()) ||
      0 != redisAppendCommand(context, expire.c_str())) {
    redis_exec_->ReleaseContext("", context, false);
    LOG_ERROR("set file stream error.(append command failed," <<
        " identity = " << identify << ")");
    return -1;
  }

  redisReply *r = NULL;
  for (int i = 0; i < 2; ++i) {
    if (0 != redisGetReply(context, (void**)&r) ||
        !(r && (r->type == REDIS_REPLY_INTEGER) && (r->integer > 0))) {
      redis_exec_->ReleaseContext("", context, false);
      LOG_ERROR("set file stream error.(get redis reply failed," <<
          " identity = " << identify << ")");
      return -1;
    }

    freeReplyObject(r);
    r = NULL;
  }

  redis_exec_->ReleaseContext("", context, true);
  return 0;
}

}
